﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

using MicroMq.Helper;

using Peer;

namespace MicroMq.Services
{
    /// <summary>
    /// 工作队列
    /// </summary>
    class EventWorkService
    {
        private readonly ConcurrentDictionary<string, Queue<WorkItem>> workItems = new ConcurrentDictionary<string, Queue<WorkItem>>();

        public void Enqueue(string topic, byte[] bytes)
        {
            var queue = workItems.GetOrAdd(topic, key => new Queue<WorkItem>());
            queue.Enqueue(new WorkItem(topic, bytes, 3));
        }
        /// <summary>
        /// 触发主题
        /// </summary>
        /// <param name="topic"></param>
        /// <param name="peerStream"></param>
        /// <returns></returns>
        public async Task Emit(string topic, PeerStream peerStream)
        {
            if (workItems.TryGetValue(topic, out var queue))
            {
                while (queue.TryDequeue(out var item))
                {
                    try
                    {
                        await peerStream.SendWithResponseAsync(item.Bytes);
                    }
                    catch (Exception e)
                    {
                        Console.WriteLine(e.Message);
                        //if (item.Reduce())
                        //{
                        // queue.Enqueue(item);
                        //}
                    }
                }
            }
        }

    }


}
